[Talend]実行するSQLを外出しにする[Java]

Talendのジョブで実行するSQLに修正があった場合、Talend起動して、ジョブを開いて対象のクエリを修正する、というのは意外と面倒な作業です(当然ながらジョブの修正に付随してビルド・デプロイ作業も発生するかと思います)。
SQLを外出しにしておけばTalendジョブの修正をするまでもなくSQLファイルの修正だけで事が済むようになりますので、手間がかからなくなるかと思います。
今回はTalendで外部SQLファイルを読み込み、MySQLに対してクエリを実行するジョブを作成するジョブを作成してみました。

実行環境

今回の実行環境は以下になります。

windows8.1

Talend 6.0.0

Java1.7

MySQL5.7

実装内容

作成したTalendのジョブは以下のようになります。

Talendでファイルの中身を文字列として取得してくれるコンポーネントがあればそれを使いたいのですが、特によさそうのがなかったので、tJavaを使いました。
大まかに説明するとtJava_1で対象のファイルの中身をSQLごとに分割します。
tMysqlConnection_1でMysqlへ接続します。
その下の、tFileList_1 →　tJava_2 → tMysqlRow_1で、分割されたSQLの読込から実行までを行います。
あとはSQLの実行結果次第でtMysqlRollback_1、tMysqlCommit_1に処理が分岐します。

各tJavaの中のコードは以下のようになっています。

try {
    // 一旦ファイルの中身を文字列として読み込む
    List<String> readLines 
        = Files.readAllLines(Paths.get(context.sqlFilePath), StandardCharsets.UTF_8);
    StringBuilder temp = new StringBuilder();
    for (String line : readLines) {
        temp.append(line + "\r\n");
    }
    // SQL文ごとにファイル分割する(tFileListで処理するため)。
    List<String> sqls = Arrays.asList(new String(temp).split(";"));
    for (String sql : sqls) {
        // 空白行は除去する
        sql = sql.trim();
        if ("".equals(sql)) {
            continue;
        }
        sql = sql + ";";
        Files.write(Paths.get(context.workFolder + System.currentTimeMillis() + ".sql"), sql.getBytes());
        // 1ミリ秒待機(同じファイル名になるのを避ける)
        Thread.sleep(1);
    }
} catch (Exception e) {
    throw e;
}

処理内容自体は簡単です。tJava_1で対象のファイルの中身を文字列として読み込み、SQLを";"で1文ごとに分割します。
SQLを1文ごとに、ファイル名を連番を付けて保存するようにします。
あとはtFileListでファイルの数だけループさせ、tJava_2で1ファイルずつファイルの中身を読み込み、tMysqlRowでクエリを実行していきます。

対象のSQLや、作業用フォルダはcontextを読むようにしていると、後々変更が簡単にできるようになるのでいいですね。
実行時にパラメータを指定する方法はこちらの記事を見て頂ければと思います。

try {
    byte[] sqlBytes = 
        Files.readAllBytes(Paths.get((String)globalMap.get("tFileList_1_CURRENT_FILEPATH")));
    String sql = new String(sqlBytes, "UTF-8");
    globalMap.put("sql", sql);
} catch (IOException e) {
    throw e;
}

tJava_2の中身はもっと簡単です。
ファイルの中身を読み込んでグローバル変数に指定しているだけです。
次のMysqlRow_1のクエリの部分に以下のように入力すればOKです。
globalMapにはObject型でsetされるので、Stringにキャストしてやる必要があります。

(String)globalMap.get("sql")

ではでは、実際に動かしてみます。
今回はTalend上から実行してみますので、contextを設定し、適当なSQLを用意して実行します。

DROP TABLE IF EXISTS test_table; 

CREATE TABLE user(id integer, name varchar (128)); 

INSERT INTO user VALUES (1, 'ユーザ1'); 

INSERT INTO user VALUES (2, 'ユーザ2'); 

うまく動いてくれました。

結果を見てみるとクエリが流れていることが確認できました。

mysql> select * from user;
+------+------------+
| id   | name       |
+------+------------+
|    1 | ユーザ1    |
|    2 | ユーザ2    |
+------+------------+
2 rows in set (0.02 sec)

今回は省いていますが、作業用フォルダに作成されたSQLは、実行後に削除する処理とか入れておくといいです。
そうしないと複数回起動した際に、過去に実行されたSQLも再実行されてしまうので。

おわりに

実行時にパラメータを変えるだけで同じジョブから違うSQLを流せるので、クエリを流すだけのジョブならば、もう新たにジョブを作成する必要もないですね。
当記事ではシンプルなクエリを単純に流してみるだけでしたが、SQLのテンプレートエンジン使ったりして、Talend内で動的にクエリを生成することもできます。
今回は以上になります。